package rocketmq

import (
	"context"
	"github.com/apache/rocketmq-clients/golang/v5"
	"shen/gotask/rocketmq/config"
	"strconv"
	"time"
)

type Producer struct{}

type ProducerMessage struct {
	Topic string   `json:"topic"`
	Body  MsgBody  `json:"body"`
	Tag   string   `json:"tag"`
	Keys  []string `json:"keys"`
}

type MsgBody struct {
	IsDelay   bool   `json:"is_delay"`
	DelayTime int    `json:"delay_time"`
	Message   string `json:"message"`
}

type MsgResult struct {
	Result  bool   `json:"result"`
	MsgId   string `json:"msgId"`
	Message any    `json:"message"`
}

func (a *Producer) SendNormalMsg(msg ProducerMessage, r *MsgResult) error {
	defer func() {
		if err := recover(); err != nil {
			StopProducer(msg.Topic)
			*r = MsgResult{
				Result:  false,
				MsgId:   "",
				Message: err,
			}
		}
	}()
	msgId, err := a.SendMsg(msg)
	if err != nil {
		return err
	}
	*r = MsgResult{
		Result:  true,
		MsgId:   msgId,
		Message: "Message sent successfully.",
	}
	return nil
}

func (a *Producer) SendMsg(message ProducerMessage) (string, error) {
	err, producer := GetProducer(message.Topic)
	if err != nil {
		return "", err
	}

	msg := &golang.Message{
		Topic: message.Topic,
	}
	msg.Body = []byte(message.Body.Message)
	if message.Body.IsDelay {
		//判断时间戳大于等于当前日期
		nowTime := time.Now().Unix()
		delayTime := int64(message.Body.DelayTime)
		msg.Body = []byte(message.Body.Message)
		var msgDelayTime time.Time
		if delayTime > nowTime {
			if differ := delayTime - nowTime; differ > config.Config.Rocketmq.RocketMqDelayMaxTime {
				msgDelayTime = time.Now().Add(time.Second * time.Duration(config.Config.Rocketmq.RocketMqDelayMaxTime))
				msg.AddProperty("ro-php-relay", strconv.FormatInt(delayTime, 10))
			} else {
				msgDelayTime = time.Now().Add(time.Second * time.Duration(differ))
				msg.AddProperty("ro-php-relay", strconv.FormatInt(delayTime, 10))
			}
		} else {
			msgDelayTime = time.Now().Add(time.Second * time.Duration(0))
		}
		msg.SetDelayTimestamp(msgDelayTime)
	}

	if len(message.Keys) > 0 {
		msg.SetKeys(message.Keys...)
	}

	if len(message.Tag) > 0 {
		msg.SetTag(message.Tag)
	}

	sendReceipts, err1 := producer.Send(context.TODO(), msg)
	if err1 != nil {
		if err.Error() != "current message type not match with topic accept message types" {
			defer StopProducer(message.Topic)
		}
		return "", err1
	}
	msgID := ""
	for _, receipt := range sendReceipts {
		msgID = receipt.MessageID
	}
	return msgID, nil
}
